package yredis

import (
	"context"
	"dgo/MyProj/PrjHttpUtils"
	"dgo/goutils/utils"
	"dgo/plg/plgnet"
	"fmt"
	"gitee.com/ymofen/supervalue"
	"github.com/go-redis/redis/v8"
	"strings"
	"time"
)

/*
*

	订阅数据, 进行输出, 只适用一个连接进行订阅
	   一般进行匹配订阅
*/
type RedisSubPlg struct {
	PlgBase

	msglst *utils.SyncCycleList

	sublst string

	redisInner               *RedisInner
	reids_sub                *redis.PubSub
	output_topic_trim_prefix string

	monitor_recv *utils.MonitorSpeedSizeRec

	disp_topic_lst     utils.GSyncMapGroup
	last_warn_disp     string
	last_recv          string
	last_recv_t        time.Time
	last_recv_workflag byte
	last_ping_msg      string
}

/*
*

	  {
	    "id": "xxx",
	    "type": "yredis.sub",
	    "conf": {
			"addr":"119.96.169.117:6379",
			"sub-channels":"RAW-RTCM-HB*,RAW-RTCM-HN*,",
			"user":"",
			"pass":"",
	        "output":"ex",
	        "ouptput-topic-trimprefix":"RAW-RTCM-"
	    }
	  }
*/
func (this *RedisSubPlg) ReloadConf(conf *wrapper.SuperValue) error {
	if conf == nil {
		return nil
	}

	this.output_topic_trim_prefix = conf.StringByName("ouptput-topic-trimprefix", "")

	this.ReloadOutputConf(conf.ValueByName("output"))

	if this.redisInner == nil {
		this.redisInner = NewRedisInner()
	}

	if this.monitor_recv == nil {
		this.monitor_recv = utils.NewMonitorSpeedSizeRec()
		utils.Monitor.AddMonitor(this.monitor_recv)
	}

	newflag := 0
	if this.redisInner.ReloadConf(conf) {
		if this.reids_sub != nil {
			this.reids_sub.Close()
			this.reids_sub = nil
		}
		this.redisInner.RequestDis("配置改变")
		this.redisInner.CheckNewClt()
		time.AfterFunc(time.Second, func() {
			err := this.redisInner.Ping()
			if err != nil {
				this.last_ping_msg = fmt.Sprintf("[%s]Reids(%s) PING err: %s", utils.NowString(), this.redisInner.RedisAddr(), err.Error())
				this.LogWarnMessage("与Redis建立连接失败, PING err: %s", err.Error())
			} else {
				this.last_ping_msg = fmt.Sprintf("[%s]Reids(%s) PING succ", utils.NowString(), this.redisInner.RedisAddr())
				this.LogInfoMessage("与Redis建立连接成功")
			}
		})
		newflag = 1
	}

	strv := conf.StringByName("sub-channels", "")
	strv = utils.ReplacePlaceholder(strv, "$$", "$$")
	if this.sublst != strv || newflag == 1 {
		ctx := context.Background()
		if len(this.sublst) > 0 && newflag == 0 {
			// 找出多订阅的
			s0 := utils.FindNonElement(strv, this.sublst, ',')
			utils.RangeStringSep(s0, ',', func(idx int, s1 string) bool {
				this.reids_sub.PUnsubscribe(ctx, s1)
				return true
			})
		}
		this.sublst = strv
		if len(strv) > 0 {
			strs := strings.Split(strv, ",")
			strs = utils.DeleteStrlstEle(strs, func(s string) bool {
				return len(utils.Trim(s)) == 0
			})
			if this.reids_sub == nil {
				this.reids_sub = this.redisInner.NewPSubscribe(strs...)
				go this.CheckStartSub(this.reids_sub)
			} else {
				this.reids_sub.PSubscribe(ctx, strs...)
			}
		}
	}

	if this.msglst == nil {
		this.msglst = utils.NewSyncCycleList()
	}

	utils.DefaultPatrolTask().AddTask(this, time.Second*30, func(id interface{}, args ...interface{}) {
		//utils.DefaultWorkers().PostTaskFunc(this, this.innerWork)
	})

	return nil
}

func (this *RedisSubPlg) CheckStartSub(sub *redis.PubSub) {
	if utils.GoFunCatchException {
		defer utils.PanicHandler()
	}

	var innerDispChannelMsg = func(msg *redis.Message) {
		this.last_recv_t = time.Now()
		this.last_recv_workflag = 1
		t0 := time.Now()
		buf := []byte(msg.Payload)
		defer func() {
			this.last_recv_workflag = 0
			ms := time.Since(t0).Milliseconds()
			if ms > 500 {
				this.last_warn_disp = fmt.Sprintf("[%s][%s]分发数据超时耗时:%d(ms), buf:%s", utils.NowString(), msg.Channel, ms, utils.BufToHexStr(buf, 0, ""))
			}
		}()
		this.last_recv = fmt.Sprintf("[%s][%s][%d]", utils.NowString(), msg.Channel, len(buf))
		this.monitor_recv.DeltaBuf(buf)
		_, topic := utils.CheckTrimPrefix(msg.Channel, this.output_topic_trim_prefix)
		this.OutputEx.DispatchData(topic, this, buf, nil)
		this.disp_topic_lst.Set(topic, time.Now())
	}

	this.LogInfoMessage("sub start...")
	for msg := range sub.Channel() {
		innerDispChannelMsg(msg)
	}
	this.LogInfoMessage("sub stop...")
}

func (this *RedisSubPlg) Close() (err error) {
	if this.redisInner != nil {
		this.redisInner.Close()
	}
	this.PlgBase.Close()
	utils.Monitor.RemoveMonitorRec(this.monitor_recv)
	utils.DefaultPatrolTask().DelTask(this)
	return
}

func (this *RedisSubPlg) OnHttpStatus(rec *plgnet.PlgHttpRec) (rval interface{}, errcode int, err error) {
	itype := rec.QryAsInt("type", 0)
	if itype == 99 {
		json := wrapper.NewSVObject()
		json.SetKeyValue("recv-info", this.monitor_recv.Info())
		json.SetKeyValue("last-recv", fmt.Sprintf("[%d]:%s", this.last_recv_workflag, this.last_recv))
		json.SetKeyValue("last-disp-warn", this.last_warn_disp)
		json.SetKeyValue("output-topic-trim-prefix", this.output_topic_trim_prefix)
		json.SetKeyValue("output", this.OutputEx.StatusString())
		json.SetKeyValue("last-ping", this.last_ping_msg)
		json.SetKeyValues("now", utils.NowString())
		if rec.QryAsInt("json", 0) == 1 {
			PrjHttpUtils.ResponseAsJSON(rec.RespW, json.AsJSONStringEx(false, true))
		} else {
			PrjHttpUtils.JSON2TableRec(rec.RespW, "", json.AsJSONString(false))
		}
	} else if itype == 98 {
		rec.ResponseSyncListNEx(this.msglst)
	} else if itype == 1 {
		sub := this.reids_sub
		if sub != nil {
			rec.ResponseText(sub.String())
		} else {
			rec.ResponseText("none")
		}
	} else if itype == 2 {
		json := wrapper.NewSVArray()
		secs := time.Duration(rec.QryAsInt("secs", 0)) * time.Second

		prefix := rec.QryAsString("prefix", "")
		trimprefix := rec.QryAsString("trimprefix", "")

		suffix := rec.QryAsString("suffix", "")
		trimsuffix := rec.QryAsString("trimsuffix", "")

		this.disp_topic_lst.Range(func(key, value interface{}) bool {
			t0 := value.(time.Time)
			if time.Since(t0) < secs {
				keystr := key.(string)
				if len(prefix) > 0 {
					if !strings.HasPrefix(keystr, prefix) {
						return true
					}
				}
				if len(trimprefix) > 0 {
					keystr = strings.TrimPrefix(keystr, trimprefix)
				}

				if len(suffix) > 0 {
					if !strings.HasSuffix(keystr, suffix) {
						return true
					}
				}
				if len(trimsuffix) > 0 {
					keystr = strings.TrimSuffix(keystr, trimsuffix)
				}

				itm := json.AppendObject()
				itm.SetKeyValue("id", keystr)
				itm.SetKeyValue("last_recv_t", utils.DateTimeString2(t0))
			}
			return true
		})
		if rec.QryAsInt("json", 0) == 1 {
			PrjHttpUtils.ResponseAsJSON(rec.RespW, json.AsJSONStringEx(false, false))
		} else {
			PrjHttpUtils.JSON2Table(rec.RespW, "", json.AsJSONString(false))
		}
	} else {
		var sb utils.BytesBuilder
		PrjHttpUtils.AppendIndexBody(&sb, "status", "status?type=99")
		PrjHttpUtils.AppendIndexBody(&sb, "msglist", "status?type=98&n=100&searchval=")
		PrjHttpUtils.AppendIndexBody(&sb, "sub-channels", "status?type=1")
		PrjHttpUtils.AppendIndexBody(&sb, "disp-channels", "status?type=2&secs=60&json=0&prefix=&trimprefix=&suffix=&trimsuffix=")
		PrjHttpUtils.ResponseBodyHtml(rec.RespW, sb.String())
	}
	return
}

func (this *RedisSubPlg) LogInfoMessage(s string, args ...interface{}) {
	this.msglst.LogMsg(s, args...)
	utils.Log.Msgf(this.PlugId, utils.LogLV_INFO, 1, s, args...)
}

func (this *RedisSubPlg) LogWarnMessage(s string, args ...interface{}) {
	this.msglst.LogMsg(s, args...)
	utils.Log.Msgf(this.PlugId, utils.LogLV_WARN, 1, s, args...)
}

func CreateRedisSubPlgFunc(args ...interface{}) (rval interface{}, err error) {
	obj := &RedisSubPlg{}

	if len(args) > 0 {
		if str, ok := args[0].(string); ok {
			obj.PlugId = str
		}
	}
	rval = obj
	return
}
